package com.abyss.source;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.io.CsvReader;
import org.apache.flink.api.java.operators.DataSource;
/**
 * @author Abyss
 * @date 2020/10/4
 * @description 从CSV中读取数据
 * 同时使用POJO类来封装数据的schema
 */
public class CSVFileSourceDemo {
    public static void main(String[] args) throws Exception {
            // 1. env
            ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

            // 2. CSV source
            // readCsvFile得到的不是source 是csvReader对象
            CsvReader csvReader = env.readCsvFile("/Users/abyss/Dev/toys/flink/H-flink-learn/src/main/resources/score.csv");
            /*
             * 参数说明：
             * fieldDelimiter设置分隔符，默认的是","
             * ignoreFirstLine忽略第一行
             * includeFields是设置选取哪几列，我这里是第四列不选取的。
             * pojoType和后面字段名，就是对应列。字段名是不需要对应原始文件header字段名，
             * 但必须与POJO例如UserInfo.class里的字段名一一对应
             */
            DataSource<Student> studentDataSource = csvReader.fieldDelimiter(",")   // 字段分隔符默认,
                    .ignoreFirstLine()              // 忽略第一行
                    .includeFields(true, true, true)// 确定哪些字段需要
                    .pojoType(Student.class, "id", "name", "address");

            studentDataSource.print();
        }

    /*
    构建一个POJO类来封装数据的schema
    1. get和set 方法
    2. 全参数的构造方法和 空参数构造方法
    3. 可选(toString)
     */
        public static class Student {
            private Integer id;
            private String name;
            private String address;

            public Student(Integer id, String name, String address) {
                this.id = id;
                this.name = name;
                this.address = address;
            }

            public Student() {}

            public Integer getId() {
                return id;
            }

            public void setId(Integer id) {
                this.id = id;
            }

            public String getName() {
                return name;
            }

            public void setName(String name) {
                this.name = name;
            }

            public String getAddress() {
                return address;
            }

            public void setAddress(String address) {
                this.address = address;
            }

            @Override
            public String toString() {
                return "Student{" +
                        "id=" + id +
                        ", name='" + name + '\'' +
                        ", address='" + address + '\'' +
                        '}';
            }

    }
}
